package com.usian.mq;

import com.usian.mapper.LocalMessageMapper;
import com.usian.pojo.LocalMessage;
import com.usian.utils.JsonUtils;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.core.RabbitTemplate.ConfirmCallback;
import org.springframework.amqp.rabbit.core.RabbitTemplate.ReturnCallback;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;

/**
 * 1.发送mq消息
 * 2.mq收到后把把本地消息表状态修改为1 ConfirmCallback接口里的confirm方法被调用也就是本类sendMsg方法里rabbitTemplate.setConfirmCallback(this);
 *   执行了说明mq消息发送成功并执行confirm此方法
 *
 */
@Component
public class MQSender implements ReturnCallback, ConfirmCallback{
    @Resource
    private LocalMessageMapper localMessageMapper;

    @Autowired
    private AmqpTemplate amqpTemplate;
    //发送MQ消息
    public void sendMsg(LocalMessage localMessage) {
        RabbitTemplate rabbitTemplate=(RabbitTemplate) this.amqpTemplate;
        rabbitTemplate.setConfirmCallback(this);//成功回调
        rabbitTemplate.setReturnCallback(this);//失败回退
        //用于确认之后更改本地消息状态或删除本地消息--本地消息id  也就是下面成功回调方法confirm用到的
        CorrelationData correlationData = new CorrelationData(localMessage.getTxNo());
        //发送消息
        rabbitTemplate.convertAndSend("order_exchange","order.add",
                JsonUtils.objectToJson(localMessage),correlationData);
    }
//成功回调
    @Override
    public void confirm(CorrelationData correlationData, boolean b, String s) {
        String id = correlationData.getId();//sendMsg里CorrelationData里的参数localMessage.getTxNo()
        if (b){
            // 消息发送成功,更新本地消息为已成功发送状态或者直接删除该本地消息记录  这里修改状态为1
            String txNo = correlationData.getId();
            LocalMessage localMessage = new LocalMessage();
            localMessage.setTxNo(txNo);
            localMessage.setState(1);
            localMessageMapper.updateByPrimaryKeySelective(localMessage);
        }

    }
//失败回退
    @Override
    public void returnedMessage(Message message, int replyCode, String replyText,
                                String exchange, String routingKey) {
        System.out.println("return--message:" + new String(message.getBody())
                + ",exchange:" + exchange + ",routingKey:" + routingKey);

    }
}
